import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashSet;
import java.util.PriorityQueue;
import java.util.Comparator;
import java.util.Map;
import java.util.HashMap;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class exp2 {

    // simple CSV splitter that respects quoted commas
    public static String[] splitCSV(String line) {
        return line.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
    }

    // ---------------- Mapper ----------------
    public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text outKey = new Text();
        private HashSet<String> stopWords = new HashSet<>();
        private boolean headerSkipped = false;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            URI[] cacheFiles = context.getCacheFiles();
            if (cacheFiles != null) {
                for (URI uri : cacheFiles) {
                    String name = uri.getPath();
                    if (name.endsWith("stop-word-list.txt")) {
                        FileSystem fs = FileSystem.get(context.getConfiguration());
                        Path p = new Path(uri);
                        try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(p)))) {
                            String line;
                            while ((line = br.readLine()) != null) {
                                line = line.trim().toLowerCase();
                                if (!line.isEmpty()) stopWords.add(line);
                            }
                        }
                    }
                }
            }
        }

        private boolean looksLikeHeader(String lineLower) {
            return lineLower.contains("text") && lineLower.contains("sentiment");
        }

        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (line == null) return;
            String trimmed = line.trim();
            if (trimmed.isEmpty()) return;

            String low = trimmed.toLowerCase();
            if (!headerSkipped && looksLikeHeader(low)) {
                headerSkipped = true;
                return;
            }

            String[] cols = splitCSV(line);
            int sentIdx = -1, textIdx = -1;
            for (int i = 0; i < cols.length; i++) {
                String c = cols[i].trim().toLowerCase();
                if (c.equals("sentiment") || c.equals("label")) sentIdx = i;
                if (c.equals("text") || c.equals("title")) textIdx = i;
            }
            if (sentIdx == -1 && cols.length >= 1) sentIdx = cols.length - 1;
            if (textIdx == -1) {
                for (int i = 0; i < cols.length; i++) {
                    if (i == sentIdx) continue;
                    if (cols[i].trim().length() > 0) { textIdx = i; break; }
                }
            }
            if (textIdx < 0 || sentIdx < 0 || textIdx >= cols.length || sentIdx >= cols.length) return;

            String text = cols[textIdx];
            String sentimentStr = cols[sentIdx].trim();
            if (sentimentStr.isEmpty()) return;

            int sentiment;
            if (sentimentStr.equals("1") || sentimentStr.equalsIgnoreCase("positive") || sentimentStr.equalsIgnoreCase("pos")) sentiment = 1;
            else if (sentimentStr.equals("-1") || sentimentStr.equalsIgnoreCase("negative") || sentimentStr.equalsIgnoreCase("neg")) sentiment = -1;
            else {
                try {
                    int s = Integer.parseInt(sentimentStr);
                    sentiment = s > 0 ? 1 : -1;
                } catch (Exception e) { return; }
            }

            String cleaned = text.toLowerCase().replaceAll("[^a-zA-Z\\s]", " ");
            StringTokenizer itr = new StringTokenizer(cleaned);
            while (itr.hasMoreTokens()) {
                String token = itr.nextToken().trim();
                if (token.isEmpty()) continue;
                if (stopWords.contains(token)) continue;
                if (token.length() <= 1) continue;
                outKey.set((sentiment == 1 ? "POS" : "NEG") + "\t" + token);
                context.write(outKey, one);
            }
        }
    }

    // ---------------- Combiner ----------------
    public static class SumCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int s = 0;
            for (IntWritable v : values) s += v.get();
            result.set(s);
            context.write(key, result);
        }
    }

    // ---------------- Reducer (job1) ----------------
    public static class SumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int s = 0;
            for (IntWritable v : values) s += v.get();
            result.set(s);
            context.write(key, result);
        }
    }

    // ---------------- Job2 Mapper ----------------
    public static class TopKMapper extends Mapper<LongWritable, Text, Text, Text> {
        private Text outKey = new Text();
        private Text outVal = new Text();
        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return;
            String[] parts = line.split("\\t");
            if (parts.length == 3) {
                outKey.set(parts[0]);
                outVal.set(parts[1] + "\t" + parts[2]);
                context.write(outKey, outVal);
            } else if (parts.length == 2) {
                // key may contain "POS\tword"
                String keyPart = parts[0];
                String count = parts[1];
                String[] kp = keyPart.split("\\t", 2);
                if (kp.length == 2) {
                    outKey.set(kp[0]);
                    outVal.set(kp[1] + "\t" + count);
                    context.write(outKey, outVal);
                }
            }
        }
    }

    // ---------------- Job2 Reducer (top-K) ----------------
    public static class TopKReducer extends Reducer<Text, Text, Text, IntWritable> {
        private MultipleOutputs<Text, IntWritable> mos;
        private int K = 100;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            mos = new MultipleOutputs<>(context);
            String kstr = context.getConfiguration().get("exp2.topk");
            if (kstr != null) {
                try { K = Integer.parseInt(kstr); } catch (Exception e) {}
            }
        }

        @Override
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            PriorityQueue<Map.Entry<String,Integer>> pq = new PriorityQueue<>(K, new Comparator<Map.Entry<String,Integer>>() {
                public int compare(Map.Entry<String,Integer> a, Map.Entry<String,Integer> b) {
                    return Integer.compare(a.getValue(), b.getValue());
                }
            });

            for (Text v : values) {
                String s = v.toString();
                String[] p = s.split("\\t");
                if (p.length < 2) continue;
                String word = p[0];
                int cnt;
                try { cnt = Integer.parseInt(p[1]); } catch (Exception e) { continue; }
                Map.Entry<String,Integer> entry = new java.util.AbstractMap.SimpleEntry<>(word, cnt);
                if (pq.size() < K) pq.add(entry);
                else if (pq.peek().getValue() < cnt) { pq.poll(); pq.add(entry); }
            }

            java.util.List<Map.Entry<String,Integer>> out = new java.util.ArrayList<>();
            while (!pq.isEmpty()) out.add(pq.poll());
            out.sort(new Comparator<Map.Entry<String,Integer>>() {
                public int compare(Map.Entry<String,Integer> a, Map.Entry<String,Integer> b) {
                    return Integer.compare(b.getValue(), a.getValue());
                }
            });

            String base = key.toString().equals("POS") ? "positive" : "negative";
            for (Map.Entry<String,Integer> e : out) {
                mos.write(base, new Text(e.getKey()), new IntWritable(e.getValue()), base + "/part");
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            if (mos != null) mos.close();
        }
    }

    // ---------------- main ----------------
    public static void main(String[] args) throws Exception {
        if (args.length < 3) {
            System.err.println("Usage: exp2 <input> <intermediate> <output> -stop <stop-word-list.txt>");
            System.exit(2);
        }

        String input = args[0];
        String interm = args[1];
        String output = args[2];
        String stopFile = "stop-word-list.txt";
        for (int i = 3; i < args.length; i++) {
            if (args[i].equals("-stop") && i + 1 < args.length) {
                stopFile = args[i+1];
                i++;
            }
        }

        Configuration conf = new Configuration();
        Job job1 = Job.getInstance(conf, "exp2-wordcount-by-sentiment");
        job1.setJarByClass(exp2.class);
        job1.addCacheFile(new URI(stopFile + "#stop-word-list.txt"));
        job1.setMapperClass(TokenizerMapper.class);
        job1.setCombinerClass(SumCombiner.class);
        job1.setReducerClass(SumReducer.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job1, new Path(input));
        FileOutputFormat.setOutputPath(job1, new Path(interm));
        boolean ok = job1.waitForCompletion(true);
        if (!ok) { System.exit(1); }

        Configuration conf2 = new Configuration();
        conf2.set("exp2.topk", "100");
        Job job2 = Job.getInstance(conf2, "exp2-topk-by-sentiment");
        job2.setJarByClass(exp2.class);
        job2.setMapperClass(TopKMapper.class);
        job2.setReducerClass(TopKReducer.class);
        job2.setMapOutputKeyClass(Text.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(IntWritable.class);
        MultipleOutputs.addNamedOutput(job2, "positive", TextOutputFormat.class, Text.class, IntWritable.class);
        MultipleOutputs.addNamedOutput(job2, "negative", TextOutputFormat.class, Text.class, IntWritable.class);
        FileInputFormat.addInputPath(job2, new Path(interm));
        FileOutputFormat.setOutputPath(job2, new Path(output));
        ok = job2.waitForCompletion(true);
        if (!ok) { System.exit(1); }

        System.out.println("exp2 finished.");
        System.exit(0);
    }
}
